package org.zjt.rxjava.lession.first;

import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.subjects.AsyncSubject;
import io.reactivex.subjects.BehaviorSubject;
import io.reactivex.subjects.PublishSubject;
import io.reactivex.subjects.ReplaySubject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.PrintStream;

/**
 * DESC
 *
 * @author
 * @create 2017-05-19 上午11:14
 **/
public class Subject {

    private static final Logger LOG = LoggerFactory.getLogger(ObservLearning.class);
    private static final PrintStream origin = System.out;
    public static void main(String[] args) {

        //
        testSubjectBridge();

    }

    /**
     *
     * AsyncSubject:
     *  Observer会接收AsyncSubject的`onComplete()之前的最后一个数据
     *
     *      只返回：asyncSubject:asyncSubject3
     *
     *      PS: onComplete()之前的最后一个数据
     */
    public static void testAsyncSubject(){
        AsyncSubject<String> asyncSubject = AsyncSubject.create();
        asyncSubject.onNext("asyncSubject1");
        asyncSubject.onNext("asyncSubject2");
        asyncSubject.onNext("asyncSubject3");
        asyncSubject.onComplete();
        asyncSubject.subscribe(new Observer<String>() {

            @Override
            public void onError(Throwable e) {

                LOG.info("asyncSubject onError");  //不输出（异常才会输出）
            }

            @Override
            public void onComplete() {
                LOG.info("asyncSubject onCompleted");  //输出 asyncSubject onCompleted
            }

            @Override
            public void onSubscribe(Disposable disposable) {

            }

            @Override
            public void onNext(String s) {

                LOG.info("asyncSubject:"+s);  //输出asyncSubject:asyncSubject3
            }
        });
    }


    /**
     * BehaviorSubject:
     *
     *      Observer会接收到BehaviorSubject被订阅之前的最后一个数据，再接收其他发射过来的数据
     *
     *      PS：onComplete()之前的最后一个数据 + onComplete()之后的所有
     */
    public static void testBehaviorSubject(){
        BehaviorSubject<String> behaviorSubject = BehaviorSubject.create();
        behaviorSubject.onNext("behaviorSubject1");
        behaviorSubject.onNext("behaviorSubject2");
        behaviorSubject.subscribe(new Observer<String>() {

            @Override
            public void onError(Throwable e) {

                LOG.info("behaviorSubject:error");
            }

            @Override
            public void onComplete() {
                LOG.info("behaviorSubject:complete");

            }

            @Override
            public void onSubscribe(Disposable disposable) {

            }

            @Override
            public void onNext(String s) {

                // TODO: 2017/5/19 得到1、3、4消息
                LOG.info("behaviorSubject:"+s);
            }
        });

        behaviorSubject.onNext("behaviorSubject3");
        behaviorSubject.onNext("behaviorSubject4");
    }


    /***
     * PublishSubject：
     *
     *      Observer只会接收到PublishSubject被订阅之后发送的数据
     *
     *      PS:只得到订阅后的数据
     */
    public static void testPublishSubject(){
        PublishSubject<String> publishSubject = PublishSubject.create();
        publishSubject.onNext("publishSubject1");
        publishSubject.onNext("publishSubject2");
        publishSubject.subscribe(new Observer<String>() {
            @Override
            public void onComplete() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onSubscribe(Disposable disposable) {

            }

            @Override
            public void onNext(String s) {
                LOG.info("publishSubject observer1:"+s);
            }
        });
        publishSubject.onNext("publishSubject3");
        publishSubject.onNext("publishSubject4");
    }


    /**
     * ReplaySubject:
     *      ReplaySubject会发射所有数据给观察者，无论它们是何时订阅的。
     *
     *      PS:得到所有信息，但可以设置缓存内容数目
     */
    public static void testReplaySubject(){
        ReplaySubject<String> replaySubject = ReplaySubject.create();

        //创建默认初始缓存容量大小为16的ReplaySubject，当数据条目超过16会重新分配内存空间，使用这种方式，不论ReplaySubject何时被订阅，Observer都能接收到数据
        //replaySubject = ReplaySubject.create(100);//创建指定初始缓存容量大小为100的ReplaySubject
        //replaySubject = ReplaySubject.createWithSize(2);//只缓存订阅前最后发送的2条数据
        //replaySubject=ReplaySubject.createWithTime(1,TimeUnit.SECONDS,Schedulers.computation());// replaySubject被订阅前的前1秒内发送的数据才能被接收


        replaySubject.onNext("replaySubject:pre1");
        replaySubject.onNext("replaySubject:pre2");
        replaySubject.onNext("replaySubject:pre3");
        replaySubject.subscribe(LOG :: info);
        replaySubject.onNext("replaySubject:after1");
        replaySubject.onNext("replaySubject:after2");
    }

    /**
     * 可以将作为Subject作为传输的桥梁
     */
    public static void testSubjectBridge(){
        PublishSubject publishSubject = PublishSubject.create();
        Observable.create(a -> {
            a.onNext("Hell,o,wor,ld");
            a.onNext("H2ell,o,wor,ld");
            a.onComplete();
            //int i =1/0;
            // TODO: 2017/5/18  onError()函数向订阅者发送一个异常。
            //a.onError(new Exception("zhangsan"));
        }).subscribe(publishSubject);

        publishSubject.subscribe(origin::println);
        origin.println("111111");
    }

}
